package com.atguigu.day03;

import com.atguigu.bean.Event;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;
import java.util.Random;

public class Flink05_Source_Custom {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //TODO 2.从自定义数据源获取数据

        DataStreamSource<Event> streamSource = env.addSource(new ClickSource()).setParallelism(2);

        streamSource.process(new ProcessFunction<Event, String>() {
            @Override
            public void processElement(Event value, ProcessFunction<Event, String>.Context ctx, Collector<String> out) throws Exception {
                System.out.println("WaterMark:" + ctx.timerService().currentWatermark());
            }
        });

        streamSource.print();

        env.execute();
    }

//    public static class ClickSource implements SourceFunction<Event>{
    //通过实现ParallelSourceFunction<OUT>这个接口可以设置多并行度
    public static class ClickSource implements ParallelSourceFunction<Event> {

        //标记，用来控制循环
        private Boolean isRunning = true;

        /**
         * 主要生成数据的方法 具体获取数据的逻辑写在这个方法中
         * @param ctx 可以将数据发送出去
         * @throws Exception
         */
        @Override
        public void run(SourceContext<Event> ctx) throws Exception {
           Random random = new Random();
            System.out.println("-----------开始生成数据------------");
            String[] users = {"Songsong", "Bingbing"};
            String[] urls = {"./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2"};

            while (isRunning){
                Event event = new Event(users[random.nextInt(users.length)], urls[random.nextInt(urls.length)], System.currentTimeMillis());

                //将数据发送出去，并指定事件时间戳
//                ctx.collectWithTimestamp(event,event.timestamp);

                //发送WaterMark
//                ctx.emitWatermark(new Watermark(event.timestamp-1));

                ctx.collect(event);

               Thread.sleep(500);
           }
        }

        /**
         * 通常在run方法中有个循环，通过这个方法来终止循环，这个方法不是我们自己调用，而是内部掉的。
         */
        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
